package cn.com.sinosoft.dbyb.mr.function;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.MultipleInputs;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.Tool;

import java.io.IOException;
import java.math.BigDecimal;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Objects;

/**
 * ClassName: CusJob
 * Description: 自定义的算法
 * Date: 2018/4/4 9:46 【需求编号】
 *
 * @author Sam Sho
 * @version V1.0.0
 */
public class CusJob extends Configured implements Tool {

    /**
     * 计算往期数据 Mapper
     */
    static class CulOldMapper extends Mapper<LongWritable, Text, Text, CusBean> {

        /**
         * 个人编号,作为一行数据的key
         */
        private Text aac999 = new Text();

        @Override
        protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
            // 获取每行数据的值
            String lineValue = value.toString().trim();
            // 进行分割
            String[] arr = lineValue.split("\\|");
            aac999.set(arr[0]);
            CusBean dataBean = new CusBean();
            dataBean.loadOldDate(arr);
            dataBean.setFlag(0);
            context.write(aac999, dataBean);
        }
    }

    /**
     * 本期计算Mapper
     */
    static class CulMapper extends Mapper<LongWritable, Text, Text, CusBean> {

        /**
         * 个人编号,作为一行数据的key
         */
        private Text aac999 = new Text();

        @Override
        protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
            // 获取每行数据的值
            String lineValue = value.toString().trim();
            // 进行分割
            String[] arr = lineValue.split("\\|");
            aac999.set(arr[0]);
            CusBean dataBean = new CusBean();
            dataBean.loadDate(arr);
            dataBean.setFlag(1);
            context.write(aac999, dataBean);
        }
    }

    /**
     * 计算Reduce
     */
    static class CulReduce extends Reducer<Text, CusBean, CusBean, NullWritable> {

        /**
         * 自负费用
         * rate1~2-普通参保人员支付比例
         */
        double[][] rate1 = {{0, 6000, 0, 800},
                {6000, 10000, 0.3, 800},
                {10000, 20000, 0.4, 2000}};

        /**
         * 普通人员-自负加合规费用比例
         */
        double[][] rate2 = {{20000, 50000, 0.5, 0},
                {50000, 100000, 0.6, 15000},
                {100000, 200000, 0.7, 45000},
                {200000, Double.MAX_VALUE, 0.8, 115000}};

        /**
         * 自负费用
         * 实时救助人员支付比例
         */
        double[][] rate3 = {{0.0000, 3000.0000, 0.0000, 800.0000},
                {3000.0000, 6000.0000, 0.3500, 800.0000},
                {6000.0000, 10000.0000, 0.3500, 1850.0000},
                {10000.0000, 20000.0000, 0.4500, 3250.0000}};

        /**
         * 实时救助人员-自负加合规费用比例
         */
        double[][] rate4 = {{20000.0000, 50000.0000, 0.5500, 0.0000},
                {50000.0000, 100000.0000, 0.6500, 16500.0000},
                {100000.0000, 200000.0000, 0.7500, 49000.0000},
                {200000.0000, Double.MAX_VALUE, 0.8500, 124000.0000}};

        /**
         * 普通参保人员: 累计自负赔付
         *
         * @param sumAkc254
         * @return
         */
        BigDecimal getRate1(double sumAkc254) {
            BigDecimal bigDecimal = new BigDecimal("0");
            if (sumAkc254 > 20000) {
                bigDecimal = new BigDecimal("6000");
            } else if (sumAkc254 < 6000) {
            } else {
                for (double[] doubles : rate1) {
                    if (sumAkc254 > doubles[0] && sumAkc254 <= doubles[1]) {
//                      sumAkc254Compensate = (sumAkc254 - doubles[0]) * doubles[2] + doubles[3];
                        bigDecimal = (BigDecimal.valueOf(sumAkc254).subtract(BigDecimal.valueOf(doubles[0]))).multiply(BigDecimal.valueOf(doubles[2])).add(BigDecimal.valueOf(doubles[3]));
                        break;
                    }
                }
            }
            return bigDecimal;
        }

        /**
         * 普通参保人员：自负+ 合规 累计赔付费用
         *
         * @param sumAkc254AddComplianceOwnExpense
         * @return
         */
        BigDecimal getRate2(double sumAkc254AddComplianceOwnExpense) {
            BigDecimal bigDecimal = new BigDecimal("0");
            // 自负+ 计算后的合规 赔付费用
            if (sumAkc254AddComplianceOwnExpense <= 20000) {
            } else {
                for (double[] doubles : rate2) {
                    if (sumAkc254AddComplianceOwnExpense > doubles[0] && sumAkc254AddComplianceOwnExpense <= doubles[1]) {
//                      umAkc254AddCOECompensate = (sumAkc254AddComplianceOwnExpense - doubles[0]) * doubles[2] + doubles[3];
                        bigDecimal = (BigDecimal.valueOf(sumAkc254AddComplianceOwnExpense).subtract(BigDecimal.valueOf(doubles[0]))).multiply(BigDecimal.valueOf(doubles[2])).add(BigDecimal.valueOf(doubles[3]));
                        break;
                    }
                }
            }
            return bigDecimal;
        }


        /**
         * 救助人员：累计自负赔付
         *
         * @param sumAkc254
         * @return
         */
        BigDecimal getRate3(double sumAkc254) {
            BigDecimal bigDecimal = new BigDecimal("0");
            if (sumAkc254 > 20000) {
                bigDecimal = new BigDecimal("7750");
            } else if (sumAkc254 < 3000) {
            } else {
                for (double[] doubles : rate3) {
                    if (sumAkc254 > doubles[0] && sumAkc254 <= doubles[1]) {
//                      sumAkc254Compensate = (sumAkc254 - doubles[0]) * doubles[2] + doubles[3];
                        bigDecimal = (BigDecimal.valueOf(sumAkc254).subtract(BigDecimal.valueOf(doubles[0]))).multiply(BigDecimal.valueOf(doubles[2])).add(BigDecimal.valueOf(doubles[3]));
                        break;
                    }
                }
            }
            return bigDecimal;
        }

        /**
         * 救助人员：自负+ 合规 累计赔付费用
         *
         * @param sumAkc254AddComplianceOwnExpense
         * @return
         */
        BigDecimal getRate4(double sumAkc254AddComplianceOwnExpense) {
            BigDecimal bigDecimal = new BigDecimal("0");
            if (sumAkc254AddComplianceOwnExpense <= 20000) {
            } else {
                for (double[] doubles : rate4) {
                    if (sumAkc254AddComplianceOwnExpense > doubles[0] && sumAkc254AddComplianceOwnExpense <= doubles[1]) {
//                      sumAkc254AddCOECompensate = (sumAkc254AddComplianceOwnExpense - doubles[0]) * doubles[2] + doubles[3];
                        bigDecimal = (BigDecimal.valueOf(sumAkc254AddComplianceOwnExpense).subtract(BigDecimal.valueOf(doubles[0]))).multiply(BigDecimal.valueOf(doubles[2])).add(BigDecimal.valueOf(doubles[3]));
                        break;
                    }
                }
            }

            return bigDecimal;
        }


        @Override
        protected void reduce(Text key, Iterable<CusBean> values, Context context) throws IOException, InterruptedException {
            ArrayList<CusBean> previousDataList = new ArrayList<>();
            ArrayList<CusBean> thisDataList = new ArrayList<>();
            // 区分往期与本次数据 0 = flag 为往期数据
            for (CusBean value : values) {
                CusBean dataBean = new CusBean();
                value.copyProperties(dataBean);
                if (0 == dataBean.getFlag()) {
                    previousDataList.add(dataBean);
                } else {
                    thisDataList.add(dataBean);
                }
            }
            Collections.sort(previousDataList);
            Collections.sort(thisDataList);


            // 往期累计自负
            BigDecimal previousSumAkc254 = new BigDecimal("0");
            // 往期累计合规自费
            BigDecimal previousSumComplianceOwnExpense = new BigDecimal("0");
            // 往期累计自负合规
            BigDecimal previousSumAkc254AddComplianceOwnExpense = new BigDecimal("0");
            // 往期累计自负赔付
            BigDecimal previousSumAkc254Compensate = new BigDecimal("0");
            // 往期累计自负加合规赔付
            BigDecimal previousSumAkc254AddCOECompensate = new BigDecimal("0");

            // 获取上个月最后一张账单的信息
            if (!previousDataList.isEmpty()) {
                // 获取上个月最后一张账单
                CusBean cusBean = previousDataList.get(previousDataList.size() - 1);

                // 累计自负
                previousSumAkc254 = new BigDecimal(cusBean.getSumAkc254());
                // 累计合规费用
                previousSumComplianceOwnExpense = new BigDecimal(cusBean.getSumComplianceOwnExpense());
                // 累计自负+合规
                previousSumAkc254AddComplianceOwnExpense = new BigDecimal(cusBean.getSumAkc254AddComplianceOwnExpense());
                // 累计累计自负赔付
                previousSumAkc254Compensate = new BigDecimal(cusBean.getSumAkc254Compensate());
                // 累计累计自负加合规赔付
                previousSumAkc254AddCOECompensate = new BigDecimal(cusBean.getSumAkc254AddCOECompensate());
            }


            // 本次累计自负赔付
            BigDecimal sumAkc254Compensate;
            // 本次累计自负加合规赔付
            BigDecimal sumAkc254AddCOECompensate;
            for (CusBean thisData : thisDataList) {

                // 往期累计自负+ 本次累计自负
                BigDecimal sumAkc254 = previousSumAkc254.add(new BigDecimal(thisData.getSumAkc254()));

                // 往期累计合规 + 本次累计合规
                BigDecimal sumComplianceOwnExpense = previousSumComplianceOwnExpense.add(new BigDecimal(thisData.getSumComplianceOwnExpense()));
                // 累计自负费用 + 累计合规费用
                BigDecimal sumAkc254AddComplianceOwnExpense = sumAkc254.add(sumComplianceOwnExpense);

                // 1-救助
                if (Objects.equals("1", thisData.getSFlag())) {
                    // 累计自负-赔付计算
                    sumAkc254Compensate = getRate3(sumAkc254.doubleValue());
                    // 自负+ 计算后的合规 赔付费用
                    sumAkc254AddCOECompensate = getRate4(sumAkc254AddComplianceOwnExpense.doubleValue());
                } else {
                    // 累计自负-赔付计算
                    sumAkc254Compensate = getRate1(sumAkc254.doubleValue());
                    // 自负+ 计算后的合规 赔付费用
                    sumAkc254AddCOECompensate = getRate2(sumAkc254AddComplianceOwnExpense.doubleValue());
                }

                // 两笔累计赔付:取两位小数
                sumAkc254Compensate = sumAkc254Compensate.setScale(2, BigDecimal.ROUND_HALF_UP);
                sumAkc254AddCOECompensate = sumAkc254AddCOECompensate.setScale(2, BigDecimal.ROUND_HALF_UP);

                // 开始计算本次赔付
                // 2-救助转普通，需要特殊处理，按照普通人员，重新计算上个月的赔付数据
                if (Objects.equals("2", thisData.getSFlag())) {

                    BigDecimal tSumAkc254Compensate = getRate1(previousSumAkc254.doubleValue());
                    BigDecimal subtract = sumAkc254Compensate.subtract(tSumAkc254Compensate);
                    thisData.setAkc254Compensate(subtract.toString());

                    // 自负+ 计算后的合规 赔付费用
                    BigDecimal tSumAkc254AddCOECompensate = getRate2(previousSumAkc254AddComplianceOwnExpense.doubleValue());
                    BigDecimal subtract1 = sumAkc254AddCOECompensate.subtract(tSumAkc254AddCOECompensate);
                    thisData.setAkc254AddCOECompensate(subtract1.toString());

                } else {
                    //非2 ，无特殊处理，只需要减去往期累计即可
                    // 本次自负赔付: 本次累计自负赔付 - 往期自负赔付
                    BigDecimal subtract = sumAkc254Compensate.subtract(previousSumAkc254Compensate);
                    thisData.setAkc254Compensate(subtract.toString());
                    // 中间赋值，下一笔账单需要减去这次的账单
                    previousSumAkc254Compensate = sumAkc254Compensate;

                    // 本次自负加合规赔付：本次累计自负加合规赔付 - 往期自负加合规赔付
                    BigDecimal subtract1 = sumAkc254AddCOECompensate.subtract(previousSumAkc254AddCOECompensate);
                    thisData.setAkc254AddCOECompensate(subtract1.toString());
                    // 中间赋值
                    previousSumAkc254AddCOECompensate = sumAkc254AddCOECompensate;
                }


                // 累计自负费用
                thisData.setSumAkc254(sumAkc254.toString());
                // 累计合规费用
                thisData.setSumComplianceOwnExpense(sumComplianceOwnExpense.toString());
                // 累计自负费用+累计合规费用
                thisData.setSumAkc254AddComplianceOwnExpense(sumAkc254AddComplianceOwnExpense.toString());

                // 累计自负赔付
                thisData.setSumAkc254Compensate(sumAkc254Compensate.toString());
                // 累计自负加合规赔付
                thisData.setSumAkc254AddCOECompensate(sumAkc254AddCOECompensate.toString());
                //获取系统当前时间
                SimpleDateFormat df = new SimpleDateFormat("yyyyMMdd");
                String tCurrentDate = df.format(System.currentTimeMillis());
                thisData.setToday(tCurrentDate);
                context.write(thisData, NullWritable.get());

            }
        }
    }


    @Override
    public int run(String[] args) throws Exception {
        Configuration conf = getConf();
        Job job = Job.getInstance(conf, CusJob.class.getSimpleName());
        job.setJarByClass(CusJob.class);

        MultipleInputs.addInputPath(job, new Path(args[0]), TextInputFormat.class, CulMapper.class);
        FileOutputFormat.setOutputPath(job, new Path(args[1]));

        //上一月的数据导入
        if (args.length > 2 && args[2] != null && !"".equals(args[2])) {
            MultipleInputs.addInputPath(job, new Path(args[2]), TextInputFormat.class, CulOldMapper.class);
        }

        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(CusBean.class);

        // reduce settings
        job.setReducerClass(CulReduce.class);
        job.setOutputKeyClass(CusBean.class);
        job.setOutputValueClass(NullWritable.class);

        boolean res = job.waitForCompletion(true);

        return res ? 0 : 1;
    }
}
